Add JitteredCronTimetable for deterministic schedule jitter#69705
Add JitteredCronTimetable for deterministic schedule jitter#69705Pebble32 wants to merge 3 commits into
Conversation
|
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
|
a11aa56 to
5f777d1
Compare
Signed-off-by: Adam <111773160+Pebble32@users.noreply.github.com>
Signed-off-by: Adam <111773160+Pebble32@users.noreply.github.com>
Signed-off-by: Adam <111773160+Pebble32@users.noreply.github.com>
5f777d1 to
b668d8d
Compare
| super().__init__(cron, timezone=timezone, interval=interval, run_immediately=run_immediately) | ||
| h = int(hashlib.md5(seed.encode()).hexdigest(), 16) | ||
| self._offset = ( | ||
| datetime.timedelta(seconds=h % int(max_jitter.total_seconds())) |
There was a problem hiding this comment.
The guard is max_jitter > timedelta(0) but the divisor is int(max_jitter.total_seconds()), so any 0 < max_jitter < 1s passes the guard and then hits modulo by zero. max_jitter=timedelta(milliseconds=500) raises ZeroDivisionError here at construction. The int() also silently drops sub-second precision, so timedelta(seconds=1.5) truncates to a 1s window and the offset can never exceed it. Flooring the guard at 1s, or dropping the int() truncation, closes both.
| max_jitter: datetime.timedelta, | ||
| ) -> None: | ||
| super().__init__(cron, timezone=timezone, interval=interval, run_immediately=run_immediately) | ||
| h = int(hashlib.md5(seed.encode()).hexdigest(), 16) |
There was a problem hiding this comment.
hashlib.md5(seed.encode()) called directly raises on FIPS-enabled Python builds. The hash here isn't security-sensitive, so airflow.utils.hashlib_wrapper.md5 (which passes usedforsecurity=False) keeps it working on FIPS, the same way serialized_dag.py and dagcode.py already do. This runs scheduler-side on every deserialize, so on a FIPS build every DAG using this timetable would fail to load.
| sharing one minute still beats every DAG firing at once). | ||
|
|
||
| All ``data_interval`` and ``logical_date`` semantics are inherited from | ||
| ``CronTriggerTimetable`` -- only the wall-clock fire time is shifted. |
There was a problem hiding this comment.
"Only the wall-clock fire time is shifted" reads as if logical_date stays pinned to the cron boundary, but it doesn't. With the default interval=0, next_dagrun_info returns DagRunInfo.interval(next_start_time, next_start_time) where next_start_time is the jittered time, so logical_date (= data_interval.start) moves by the offset too. The max_jitter caveat further down ("the shift cannot push a run's logical_date across a day or period boundary") already assumes this, so the two statements contradict each other. Worth rewording to say the offset shifts logical_date/data_interval as well, or deciding whether logical_date should stay pinned to the boundary.
| ``run_immediately``. | ||
| """ | ||
|
|
||
| seed: str = attrs.field(kw_only=True, default="") |
There was a problem hiding this comment.
With seed defaulting to "", a user who sets max_jitter but forgets seed gets the same md5("") offset for every DAG, so they all fire at the same shifted instant -- the herd just moves off the cron boundary instead of spreading out, silently defeating the point of the feature. The no-op default only makes sense when max_jitter=0 too. Could this raise (or warn) when max_jitter > 0 and seed == ""? Note the core __init__ already makes seed required, so the two classes disagree on this today.
2 participants
Add
JitteredCronTimetable, an opt-inCronTriggerTimetablesubclass that shifts each DAG's fire time by a deterministic, per-DAG offset drawn from[0, max_jitter). This spreads out DAGs that share a cron expression so they no longer all fire at the same instant, without changinglogical_date/data_intervalsemantics.Why
@dailyexpands to0 0 * * *, so every daily DAG in a deployment is scheduled at exactly midnight. In large deployments this "thundering herd" at the cron boundary overloads the scheduler and workers — enough to cause task failures when dozens of DAGs are born at the same instant.The existing ways to deal with this don't actually de-collide the schedule:
@dailyintentparallelism,max_active_tasks_per_dag, pool slots)JitteredCronTimetableis the only approach that moves the fire times themselves, deterministically: the sameseed(e.g. the DAG id) always maps to the same offset, so runs stay stable and predictable across scheduler restarts and timetable serialization. Motivated by the discussion in #69027.What
JitteredCronTimetable(CronTriggerTimetable)in both the Task SDK (author-facing, attrs-based) and airflow-core (scheduler-side), plus the serialization wiring (BUILTIN_TIMETABLESmapping,serialize/deserialize, encode/decode across the SDK↔core boundary).CronTriggerTimetable:seed: strandmax_jitter: timedelta.md5(seed) % max_jitter.total_seconds(), applied as a "strip → cron → apply" coordinate shift so cron/DST alignment is fully delegated to the parent and only the wall-clock fire time is shifted.seed="",max_jitter=timedelta(0)) the offset is zero and it behaves identically toCronTriggerTimetable. Nothing changes for anyone who doesn't use it.Tests
airflow-core/tests/unit/timetables/test_jittered_cron_timetable.py, modeled on the existingtest_trigger_timetable.py:America/New_York);max_jitterreproducesCronTriggerTimetableexactly (catchup on/off);[0, max_jitter), and spread across distinct seeds;serialize/deserializeround-trips seed + window + derived offset;Was generative AI tooling used to co-author this PR?
Generated-by: Claude (Claude Code), following the guidelines
related: #69027